package main

import (
	"flag"
	"fmt"
	"sync"
	"time"
	MQTT "github.com/eclipse/paho.mqtt.golang"
)

var f MQTT.MessageHandler = func(client MQTT.Client, message MQTT.Message) {
	fmt.Printf("Topic: %s\n", message.Topic())
	fmt.Printf("Msg: %s\n", message.Payload())
}

var fail_num int = 0
var PASSWRORD string = "eyJ0eXAiOiJKV1QiLCJhbGciOiJIUzI1NiJ9.eyJpc3MiOiJ0YW8xLmppYW5nIiwidXNlciI6InRvdSJ9._yv05gR0dAFJeDoEPI8Wo5qB01Gf-cM8_M1SbdoV9jQ"

func main() {
	c:=flag.Int("n", 3000, "client nums")
	flag.Parse()
	nums := int(*c)
	wg := sync.WaitGroup{}
	for i := 0; i < nums; i++ {
		wg.Add(1)
		time.Sleep(5*time.Millisecond)
		go createTask(i, &wg)
	}

	wg.Wait()
}

func createTask(taskId int, wg *sync.WaitGroup) {
	defer wg.Done()
	opts:=MQTT.NewClientOptions().AddBroker("tcp://newmqtt.mymlsoft.com:1883").
		SetUsername("MqttTest").SetPassword(PASSWRORD)
	opts.SetClientID(fmt.Sprintf("mqttbench:%d-%d", taskId, time.Now().Unix()))
	opts.SetDefaultPublishHandler(f)
	opts.SetConnectTimeout(time.Duration(60) * time.Second)
	opts.SetAutoReconnect(true)

	c:= MQTT.NewClient(opts)
	if token := c.Connect();token.WaitTimeout(time.Duration(60)*time.Second) && token.Wait() && token.Error() != nil{
		fail_num++
		fmt.Printf("taskId:%d, fail_num:%d, error:%s\n", taskId, fail_num, token.Error())
		return
	}

	i := 0
	for{
		i ++
		time.Sleep(time.Duration(5) * time.Second)
		text := fmt.Sprintf("This is msg #%d! from task:%d", i, taskId)
		token := c.Publish("test", 1, false, text)
		token.Wait()
	}
	c.Disconnect(250)
	fmt.Println("task ok!")
}
